package thread

import (
	"fmt"
	"strings"
	"sync"
	"time"

	"gitee.com/sansaniot/ssiot-core/facade/env"
	"gitee.com/sansaniot/ssiot-core/logger"

	"ssdevice/common/constant"
	"ssdevice/common/utils"
	"ssdevice/internal/device/models"
)

type DevHisDataThread struct {
}

var isPartition = false
var partType string
var hisTableName = models.SysHistorySensor{}.TableName()
var idxLock sync.Mutex

func (DevHisDataThread) Run(threadName string) {
	logger.Infof("启动线程 - %s.", threadName)

	// 检测历史表类型(防止配置文件错误配置，以实际表信息为准)
	checkPart()

	// 初始化子线程个数
	constant.HisDataBucket = make([]*constant.CQueue, constant.DataRoutineChild)
	for i := 0; i < constant.DataRoutineChild; i++ {
		constant.HisDataBucket[i] = constant.NewCQueue()
		go func(base int) {
			child := fmt.Sprintf("%s-%d", threadName, base)
			logger.Infof("启动子线程 - %s", child)
			q := constant.HisDataBucket[base]
			dataList := make([]models.SysHistorySensor, 0)
			for {
				time.Sleep(1 * time.Second)
				for {
					s := q.Dequeue()
					if s == nil {
						break
					}
					dataList = append(dataList, s.(models.SysHistorySensor))
					if len(dataList) > constant.ThreadBatchSize {
						// 批量处理数据
						dealDevHistoryData(dataList, child)
						// 清空(减少扩容)
						dataList = dataList[0:0]
					}
				}
				if len(dataList) > 0 {
					// 批量处理数据
					dealDevHistoryData(dataList, child)
					// 清空(减少扩容)
					dataList = dataList[0:0]
				}
			}
		}(i)
	}
}

func dealDevHistoryData(dataList []models.SysHistorySensor, child string) {
	t1 := time.Now()
	//if err := env.Db.Table(hisTableName).Create(&dataList).Error; err != nil {
	//	logger.Error(err)
	//}
	sql := "INSERT INTO " + hisTableName + " (dev_sn, value, update_time, date_time) VALUES "
	valueSql := make([]string, 0)
	for _, s := range dataList {
		if isPartition {
			autoChildPartTable(s.DevSn, s.DateTime)
		}
		valueSql = append(valueSql, "('"+s.DevSn+"','"+string(s.Value)+"'::jsonb,'"+s.UpdateTimeInsStr+"','"+s.DateTime+"')")
	}
	sql += strings.Join(valueSql, ",")
	if err := env.Db.Raw(sql).Scan(&scan).Error; err != nil {
		logger.Error(err)
	}
	logger.Infof("%s sensor历史记录批量入库%d. 耗时%v", child, len(dataList), time.Since(t1).Round(time.Millisecond))
}

func checkPart() {
	if err := env.Db.Raw("SELECT relkind FROM pg_class WHERE relname = ?", hisTableName).Scan(&scan).Error; err != nil {
		logger.Error(err)
		return
	}
	if len(scan) == 0 {
		return
	}
	isPartition = scan["relkind"] == "p"
	if !isPartition {
		return
	}
	partType = env.ConfigOperator.GetString("service.device.parttype")
	relnms := make([]map[string]interface{}, 0)
	sql0 := `SELECT relname FROM pg_class WHERE relname like '` + hisTableName + `%' and relpartbound is not null`
	if err := env.Db.Raw(sql0).Scan(&relnms).Error; err != nil {
		logger.Error(err)
		return
	}
	tableNameLen := len(hisTableName)
	for _, m := range relnms {
		relname := utils.GetStringValueFiled(m, "relname")
		if partType == "date" {
			dateStr := strings.ReplaceAll(relname[tableNameLen+1:], "_", "-")
			constant.PartitionHistoryTbs[dateStr] = hisTableName
		} else {
			constant.PartitionHistoryTbs[relname[tableNameLen+1:]] = hisTableName
		}
	}
}

func autoChildPartTable(sn, datetimeStr string) {
	if partType == "date" {
		if _, ok := constant.PartitionHistoryTbs[datetimeStr]; ok {
			return
		}
		idxLock.Lock()
		defer idxLock.Unlock()
		if _, ok := constant.PartitionHistoryTbs[datetimeStr]; ok {
			return
		}
		t1 := time.Now()
		// 同建表的逻辑
		timeTime, _ := time.ParseInLocation("2006-01-02", datetimeStr, time.Local)
		timeStr2 := timeTime.AddDate(0, 0, 1).Format("2006-01-02")
		partitionRelname := hisTableName + "_" + strings.ReplaceAll(datetimeStr, "-", "_")
		if err := env.Db.Raw("create table if not EXISTS " + partitionRelname + " partition of " + hisTableName + " for values from ('" + datetimeStr + "') to ('" + timeStr2 + "')").Scan(&scan).Error; err != nil {
			logger.Error(err)
			return
		}
		env.Db.Raw("drop index if EXISTS idx_" + partitionRelname + "_sn").Scan(&scan)
		cidx := `CREATE INDEX "idx_` + partitionRelname + `_sn" ON "` + partitionRelname + `" USING btree (
					"dev_sn" COLLATE "pg_catalog"."default" "pg_catalog"."text_ops" ASC NULLS LAST
				);`
		if err := env.Db.Raw(cidx).Scan(&scan).Error; err != nil {
			logger.Error(err)
			return
		}
		logger.Infof("创建历史分区子表%s. 耗时%v", partitionRelname, time.Since(t1).Round(time.Millisecond))
		constant.PartitionHistoryTbs[datetimeStr] = hisTableName
	} else {
		snn := strings.ReplaceAll(strings.ReplaceAll(strings.ToLower(sn), "-", "_"), " ", "_")
		if _, ok := constant.PartitionHistoryTbs[snn]; ok {
			return
		}
		idxLock.Lock()
		defer idxLock.Unlock()
		if _, ok := constant.PartitionHistoryTbs[snn]; ok {
			return
		}
		// 大量设备自动创建时 延迟创建索引
		for constant.InitializingDev.Has(sn) {
			time.Sleep(3 * time.Millisecond)
		}
		t1 := time.Now()
		// 同建表的逻辑
		partitionRelname := hisTableName + "_" + snn
		if err := env.Db.Raw("create table if not EXISTS " + partitionRelname + " partition of " + hisTableName + " for values in ('" + sn + "') ").Scan(&scan).Error; err != nil {
			logger.Error(err)
			return
		}
		env.Db.Raw("drop index if EXISTS idx_" + partitionRelname + "_datetime").Scan(&scan)
		cidx := `CREATE INDEX "idx_` + partitionRelname + `_datetime" ON "` + partitionRelname + `" USING btree (
					"date_time" COLLATE "pg_catalog"."default" "pg_catalog"."text_ops" ASC NULLS LAST
				);`
		if err := env.Db.Raw(cidx).Scan(&scan).Error; err != nil {
			logger.Error(err)
			return
		}
		logger.Infof("创建历史分区子表%s. 耗时%v", partitionRelname, time.Since(t1).Round(time.Millisecond))
		constant.PartitionHistoryTbs[snn] = hisTableName
	}

}
